package com.tc.vms;

import com.tc.vms.exception.VmsClientErrorCode;
import com.tc.vms.exception.VmsClientException;
import com.tc.vms.iface.IContext;
import com.tc.vms.iface.IPublisher;
import com.tc.vms.iface.ISubscriber;
import com.tc.vms.iface.VmsServiceAdaptor;
import com.tc.vms.utils.Utils;

import java.util.Properties;

import static com.tc.vms.config.VmsClientContext.VmsClientContextBuilder;

/**
 * {@code VMSClient} supports kafka,rabbitmq original client so far. It
 * starts connection to broker while publishing or subscription in default. It
 * depends on config center to bootstrap,which should be guaranteed ping-able to
 * <em>GD6-cfgcenter-ZK-001.idc.vip.com，GD6-cfgcenter-ZK-002.idc.vip.com，GD6-cfgcenter-ZK-003.idc.vip.com</em>
 *
 * @since vmsclient v1.0.0
 */
public final class VMSClient extends VmsServiceAdaptor {
    private VmsBootstrap bootstrap;
    private final VmsClientContextBuilder builder = new VmsClientContextBuilder();

    public VMSClient() {
    }

    /**
     * @return
     */
    public VmsClientContextBuilder config() {
        return this.builder;
    }

    public <T extends IPublisher> T newPublisher(String channelName) throws VmsClientException {
        return newPublisher(channelName, null, null);
    }

    public <T extends IPublisher> T newPublisher(String channelName, String clientId) throws VmsClientException {
        return newPublisher(channelName, clientId, null);
    }

    public <T extends IPublisher> T newPublisher(String channelName, String clientId, Properties props) throws VmsClientException {
        if (!isOpen)
            throw new VmsClientException("Create Publisher failed. Maybe VmsClient service doesn't start.", VmsClientErrorCode.VMS_SYS_ERR_NOT_RUNNING);
        return bootstrap.getVmsConManager().createPublisher(channelName, clientId, props);
    }

    public <T extends ISubscriber> T newSubscriber(String queueName) throws VmsClientException {
        return newSubscriber(queueName, null, null);
    }

    public <T extends ISubscriber> T newSubscriber(String queueName, String clientId) throws VmsClientException {
        return newSubscriber(queueName, clientId, null);
    }

    public <T extends ISubscriber> T newSubscriber(String queueName, String clientId, Properties props) throws VmsClientException {
        if (!isOpen)
            throw new VmsClientException("Create Subscriber failed. Maybe VmsClient service doesn't start.", VmsClientErrorCode.VMS_SYS_ERR_NOT_RUNNING);
        return bootstrap.getVmsConManager().createSubscriber(queueName, clientId, props);
    }

    public <T extends ISubscriber> T newDLQSubscriber(String queueName) throws VmsClientException {
        if (!isOpen)
            throw new VmsClientException("Create DLQSubscriber failed. Maybe service doesn't start.", VmsClientErrorCode.VMS_SYS_ERR_NOT_RUNNING);
        String clientId = "DLQ_Subscriber_" + Utils.randomStr();
        return bootstrap.getVmsConManager().createDLQSubscriber(queueName, clientId, null);
    }

    @Override
    public boolean init(IContext context) throws VmsClientException {
        if (!isOpen) {
            bootstrap = new VmsBootstrap(builder.build());
            bootstrap.init(context);
            isOpen = true;
        }
        return isOpen;
    }

    @Override
    public boolean start() throws VmsClientException {
        return init(null);
    }

    @Override
    public void shutdown() {
        if (isOpen) {
            isOpen = false;
            bootstrap.shutdown();
            bootstrap = null;
        }
    }

    /**
     * @return Vmsclient version
     */
    public static String getVmsVersion() {
        return VmsVersion.getDescByVer(VmsVersion.CurVersion);
    }
}
